package main.scala.demo

import java.sql.{Connection, DriverManager, PreparedStatement}

import org.apache.spark.rdd.{JdbcRDD, RDD}
import org.apache.spark.{SparkConf, SparkContext}

/**
  * ActiveWebRDD
  *
  * @author zhangyimin
  *  2018-10-08 下午4:18
  * @version 1.0
  */
object ActiveWebRDD2Mysql {

  def main(args: Array[String]): Unit = {
    //配置
    val conf = new SparkConf()
    conf.setAppName("webCount")
    conf.setMaster("local")
    //创建一个sparkContext
    val sc = new SparkContext(conf)


    val rdd1 = sc.textFile("hdfs://10.16.7.36:9000/data/input/localhost_access_log.2017-07-30.txt")

    //    rdd1.map(x=>(x.substring(x.indexOf("\"")+1,x.lastIndexOf("\"")).split(" ")(1),1))
    //      .reduceByKey(_+_)
    //      .saveAsTextFile("hdfs://10.16.7.36:9000/data/output/spark/100801")
    //192.168.88.1 - - [30/Jul/2017:12:54:42 +0800] "GET /MyDemoWeb/oracle.jsp HTTP/1.1" 200 242
    val rdd2 = rdd1.map {
      line => {
        val index1 = line.indexOf("\"") + 1
        val index2 = line.lastIndexOf("\"")
        val content = line.substring(index1, index2)

        val contents = content.split(" ")
        val jspContent = contents(1)

        val jspIndex = jspContent.lastIndexOf("/")
        val jspPage = jspContent.substring(jspIndex + 1)


        (jspPage, 1)
      }
    }
    val rdd3 = rdd2.reduceByKey(_ + _)
    //排序
    val rdd4 = rdd3.sortBy(_._2, false)
    //取出前两个
    //    println(rdd4.take(2).toBuffer)
//    rdd4.saveAsTextFile("hdfs://10.16.7.36:9000/data/output/spark/100804")

    rdd4.foreachPartition(save2MySQL)

    //针对分区,创建conn,


    JdbcRDD

    sc.stop()


  }


  def save2MySQL(rdd4: Iterator[(String, Int)]) = {
    var conn: Connection =null
    var pst: PreparedStatement =null
    try {
      conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/hive_etl", "root", "123456")
      rdd4.foreach(f => {
        pst = conn.prepareCall("insert into rdd_demo_web_data values(?,?)")
        pst.setString(1, f._1)
        pst.setInt(2, f._2)
        pst.executeUpdate()
      })
    } catch {
      case e1: Exception => e1.printStackTrace()
    } finally {
      if (pst!=null){
        pst.close()
      }
      if (conn!=null) {
        conn.close()
      }
    }
  }


}





